package io.kimmking.rpcfx.register.center.zookeeper;

import com.alibaba.fastjson.JSON;
import io.kimmking.rpcfx.register.center.api.ProviderInfo;
import io.kimmking.rpcfx.register.center.api.Register;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;


/**
 * @author lfy
 */
public class ZookeeperRegister implements Register {


    private CuratorFramework client;
    private final String zookeeperServer;
    private final String namespace;

    public ZookeeperRegister(String server, String namespace) {
        this.zookeeperServer = server;
        this.namespace = namespace;
        init();
    }
    public CuratorFramework getClient() {
        return client;
    }
    private void init() {
        // start zk client
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder().connectString(zookeeperServer)
                .namespace(namespace).retryPolicy(retryPolicy).build();
        client.start();
    }
    @Override
    public void register(ProviderInfo providerInfo) {

        String serviceName = providerInfo.getServiceName();
        try {
            if ( null == client.checkExists().forPath("/" + serviceName)) {
                client.create().withMode(CreateMode.PERSISTENT).forPath("/" + serviceName, "service".getBytes());
            }
            client.create().withMode(CreateMode.EPHEMERAL).
                    forPath(generateNodeName(providerInfo, serviceName),
                            JSON.toJSONString(providerInfo).getBytes()
                    );

        } catch (Exception ex) {
            ex.printStackTrace();
            throw new RuntimeException(ex);
        }

    }


    private String generateNodeName(ProviderInfo providerInfo, String serviceName) {
        return "/" + serviceName + "/" + providerInfo.getHost() + ":" + providerInfo.getPort();
    }

}
